package ysgrpc

import (
	"context"
	"fmt"
	"gitee.com/kmyss/gf-ex/ysgrpc/proto"
	"gitee.com/kmyss/gf-ex/yslog"
	"github.com/gogf/gf/v2/errors/gcode"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/os/gfile"
	"github.com/gogf/gf/v2/os/glog"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"math"
)

type MQCfg struct {
	Host    string
	PubSize int
}

// 获取 MQCfg 连接配置参数
func getMQGRPCConnCfg(ctx context.Context) *MQCfg {
	host := FlagHost(ctx).GetVar()
	pubSize := FlagPubSize(ctx).GetVar()

	return &MQCfg{
		Host:    host.String(),
		PubSize: int(gfile.StrToSize(pubSize.String())),
	}
}

// NewMQClient 构建 YS-MQ 的 GRPC 连接客户端
func NewMQClient(ctx context.Context) *MQClient {
	m := &MQClient{
		ctx: ctx,
		cfg: getMQGRPCConnCfg(ctx),
	}

	m.init()
	return m
}

type MQClient struct {
	MQ proto.MQClient

	l    *glog.Logger // 内部日志，使用 log 函数调用
	ctx  context.Context
	cfg  *MQCfg
	conn *grpc.ClientConn
}

func (m *MQClient) log() *glog.Logger {
	if m.l == nil {
		m.l = yslog.Log(GenSubCfgNodeName("mq.client"))
	}
	return m.l
}

func (m *MQClient) init() {
	conn, err := grpc.Dial(m.cfg.Host, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		conn = nil
		m.log().Fatal(m.ctx, "建立 GRPC 连接失败！", err)
	}
	m.conn = conn

	m.MQ = proto.NewMQClient(m.conn)
}

func (m *MQClient) Sub(topics string) (proto.MQ_SubClient, error) {
	client, err := m.MQ.Sub(m.ctx, &proto.SubRequest{
		Topics: topics,
	})

	if err != nil {
		m.log().Error(m.ctx, "订阅失败！", err)
		return nil, gerror.NewCode(gcode.CodeInternalError, "订阅失败")
	}

	return client, nil
}

func (m *MQClient) Pub(topic string, message []byte) (int32, error) {

	totalNum := int32(0)
	packs := m.genPacks(topic, message)
	for i := range packs {
		m.log().Debugf(m.ctx, "当前发送：%s", packs[i].Topic)
		num, err := m.doPub(&packs[i])
		if err != nil {
			return 0, err
		}
		totalNum += num
	}

	return totalNum, nil
}

func (m *MQClient) genPacks(topic string, message []byte) (out []proto.Message) {
	dataLen := len(message)
	if dataLen > m.cfg.PubSize {
		packs := int(math.Ceil(float64(dataLen) / float64(m.cfg.PubSize)))
		m.log().Debugf(m.ctx, "消息大小：%d 分包数：%d", dataLen, packs)
		for i := 0; i < packs; i++ {
			start := i * m.cfg.PubSize     // 起始位置
			end := (i + 1) * m.cfg.PubSize // 结束位置
			if end > dataLen {
				end = dataLen
			}

			msg := message[start:end]
			packTopic := fmt.Sprintf("%s[%d-%d]", topic, i, packs)
			out = append(out, proto.Message{
				Topic:   packTopic,
				Message: msg,
			})
		}
	} else {
		out = append(out, proto.Message{
			Topic:   topic,
			Message: message,
		})
	}

	return
}

func (m *MQClient) doPub(message *proto.Message) (int32, error) {

	// 对数据做拆分
	replay, err := m.MQ.Pub(m.ctx, message)

	if err != nil {
		m.log().Errorf(m.ctx, "数据发布失败！[%s] %s", message.Topic, err)
		return 0, gerror.NewCode(gcode.CodeInternalError, "数据发布失败")
	}

	return replay.GetNum(), nil
}

func (m *MQClient) UnSub(id int32) error {
	_, err := m.MQ.UnSub(m.ctx, &proto.UnSubRequest{Id: id})
	if err != nil {
		m.log().Error(m.ctx, "取消订阅失败！", err)
		return gerror.NewCode(gcode.CodeInternalError, "取消订阅失败")
	}
	return nil
}

func (m *MQClient) Close() error {
	return m.conn.Close()
}
